package com.itbaizhan.shopping_seckill_service.service;

import cn.hutool.bloomfilter.BitMapBloomFilter;
import com.alibaba.csp.sentinel.annotation.SentinelResource;
import com.alibaba.csp.sentinel.slots.block.BlockException;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.core.toolkit.IdWorker;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.itbaizhan.shopping_common.pojo.CartGoods;
import com.itbaizhan.shopping_common.pojo.Orders;
import com.itbaizhan.shopping_common.pojo.SeckillGoods;
import com.itbaizhan.shopping_common.result.BusException;
import com.itbaizhan.shopping_common.result.CodeEnum;
import com.itbaizhan.shopping_common.service.SeckillService;
import com.itbaizhan.shopping_seckill_service.mapper.SeckillGoodsMapper;
import com.itbaizhan.shopping_seckill_service.redis.RedissonLock;
import org.apache.dubbo.config.annotation.DubboService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.serializer.StringRedisSerializer;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;
import java.math.BigDecimal;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.List;
import java.util.concurrent.TimeUnit;

/**
 * 秒杀服务实现
 */
@DubboService
@Service
public class SeckillServiceImpl implements SeckillService {
	
	@Autowired
	private SeckillGoodsMapper seckillGoodsMapper;
	
	@Autowired
	private RedisTemplate redisTemplate;
	
	@Autowired
	private BitMapBloomFilter bitMapBloomFilter;
	
	@Autowired
	private RedissonLock redissonLock;
	
	private static final String SECKILL_GOODS = "seckillGoods";// redis中秒杀商品的key
	
	
	/**
	 * 同步商品数据
	 * 每分钟查询一次数据库，更新redis中的秒杀商品数据
	 * 条件为startTime < 当前时间 < endTime，库存大于0
	 */
	@Scheduled(cron = "0 * * * * *")
	public void refreshRedis() {
		System.out.println("同步mysql秒杀商品到redis...");
		
		// 将redis中秒杀商品的库存数据同步到mysql
		List<SeckillGoods> seckillGoodsListOld = redisTemplate.boundHashOps(SECKILL_GOODS).values();
		for (SeckillGoods seckillGoods : seckillGoodsListOld) {
			// 在数据库查询秒杀商品
			QueryWrapper<SeckillGoods> queryWrapper = new QueryWrapper();
			queryWrapper.eq("goodsId", seckillGoods.getGoodsId());
			SeckillGoods sqlSeckillGoods = seckillGoodsMapper.selectOne(queryWrapper);
			if (sqlSeckillGoods != null) {
				// 修改数据库中秒杀商品的库存，和redis中的库存保持一致
				sqlSeckillGoods.setStockCount(seckillGoods.getStockCount());
				seckillGoodsMapper.updateById(sqlSeckillGoods);
			}
		}
		
		
		//1.查询数据库中正在秒杀的商品
		QueryWrapper<SeckillGoods> wrapper = new QueryWrapper<>();
		Date date = new Date();
		String now = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(date);
		wrapper.le("startTime", now)//当前时间晚于开始时间
				.ge("endTime", now)//当前时间早于结束时间
				.gt("stockCount", 0);//库存大于0
		List<SeckillGoods> seckillGoods = seckillGoodsMapper.selectList(wrapper);
		
		//2.删除redis中正在秒杀的商品
		redisTemplate.delete(SECKILL_GOODS);
		
		//3.正在秒杀中的商品添加到redis中
		for (SeckillGoods seckillGood : seckillGoods) {
			redisTemplate.boundHashOps(SECKILL_GOODS).put(seckillGood.getGoodsId(), seckillGood);
			// 将正在秒杀的商品保存到布隆过滤器
			bitMapBloomFilter.add(seckillGood.getGoodsId().toString());
		}
	}
	
	/**
	 * 前台用户查询秒杀商品
	 *
	 * @param page 页数
	 * @param size 每页条数
	 * @return 查询结果
	 */
	public Page<SeckillGoods> findPageByRedis(int page, int size) {
		//1.查询redis中秒杀商品数据
		List<SeckillGoods> seckillGoodsList = redisTemplate.boundHashOps(SECKILL_GOODS).values();
		//2.获取当前页面列表
		int start = (page - 1) * size;
		int end = start + size > seckillGoodsList.size() ? seckillGoodsList.size() : start + size;
		
		//截取当前页数结果集
		List<SeckillGoods> seckillGoods = seckillGoodsList.subList(start, end);
		
		//3.封装分页结果
		Page<SeckillGoods> seckillGoodsPage = new Page<>();
		seckillGoodsPage.setCurrent(page)//当前页
				.setSize(size)//每页条数
				.setTotal(seckillGoodsList.size())//总条数
				.setRecords(seckillGoods);//商品集合结果集
		
		return seckillGoodsPage;
	}
	
	/**
	 * 优化查询秒杀商品详情1j
	 *1先查询布隆过滤器查询
	 * 2从redis中查询秒杀商品
	 *
	 * @param goodsId 秒杀商品对应的商品Id
	 * @return 查询结果
	 */
	public SeckillGoods findSeckillGoodsByRedis(Long goodsId) {
		
		//先查询布隆过滤器查询该商品是否存在
		if (!bitMapBloomFilter.contains(goodsId.toString())) {
			System.out.println("布隆过滤器中不存在该商品");
			throw new BusException(CodeEnum.SECKILL_GOODS_NOT_EXIST);
		}
		
		// 1.从redis中查询秒杀商品
		SeckillGoods seckillGoods = (SeckillGoods) redisTemplate.boundHashOps(SECKILL_GOODS).get(goodsId);
		// 2.如果查到商品，返回
		if (seckillGoods != null) {
			System.out.println("从redis中查询秒杀商品");
			return seckillGoods;
		}
		
		return null;
	}
	
	/**
	 * 从mysql中查询秒杀商品
	 * @SentinelResource 主要用于实现流量控制、熔断降级和系统保护等功能。
	 * mySqlBlockHandler服务降级方法
	 * @param goodsId
	 * @return
	 */
	@SentinelResource(value = "findSecillGoodsByMySql",blockHandler = "mysqlBlockHandler")
	public SeckillGoods findSecillGoodsByMySql(Long goodsId) {
		
		// 3.如果没有查到商品，从数据库查询秒杀商品
		QueryWrapper<SeckillGoods> queryWrapper = new QueryWrapper();
		queryWrapper.eq("goodsId",goodsId);
		SeckillGoods seckillGoodsMysql = seckillGoodsMapper.selectOne(queryWrapper);
		System.out.println("从mysql中查询秒杀商品");
		// 4.如果该商品不在秒杀状态，抛出异常
		Date now = new Date();
		if (seckillGoodsMysql == null
				|| now.after(seckillGoodsMysql.getEndTime())
				|| now.before(seckillGoodsMysql.getStartTime())
				|| seckillGoodsMysql.getStockCount() <=0
		) {
			return null;
		}
		// 5.如果该商品在秒杀状态，将商品保存到redis，并返回该商品
		addRedisSeckillGoods(seckillGoodsMysql);
		return seckillGoodsMysql;
	}
	
	/**
	 * 降级处理
	 * @return 空值
	 */
	public SeckillGoods mysqlBlockHandler(Long goodsId, BlockException e){
		System.out.println("服务降级处理");
		return null;
	}
	
	/**
	 * 生成秒杀订单
	 *
	 * @param orders 订单数据
	 * @return
	 */
	public Orders createOrder(Orders orders) {
		
		String lockKey = orders.getCartGoods().get(0).getGoodId().toString();
		/**
		 * 添加分布式锁
		 */
		if (redissonLock.lock(lockKey, 1000 * 5)) {
			
			try {
				// 1.生成订单对象
				orders.setId(IdWorker.getIdStr()); // 手动雪花算法生产订单id
				orders.setStatus(1); // 订单状态未付款
				orders.setCreateTime(new Date()); // 订单创建时间
				orders.setExpire(new Date(new Date().getTime() + 1000 * 60 * 5));// 订单过期时间
				
				// 计算商品价格
				CartGoods cartGoods = orders.getCartGoods().get(0);
				BigDecimal price = cartGoods.getPrice();
				Integer num = cartGoods.getNum();
				BigDecimal sum = price.multiply(BigDecimal.valueOf(num));
				orders.setPayment(sum);
				
				//2.查询秒杀商品库存
				SeckillGoods seckillGoods = findSeckillGoodsByRedis(cartGoods.getGoodId());
				
				if (seckillGoods.getStockCount() <= 0) {
					throw new BusException(CodeEnum.NO_STOCK_ERROR);
				}
				
				//3.减库存,存入redis
				seckillGoods.setStockCount(seckillGoods.getStockCount() - num);
				redisTemplate.boundHashOps(SECKILL_GOODS).put(seckillGoods.getGoodsId(), seckillGoods);
				
				// 4.保存订单
				redisTemplate.setKeySerializer(new StringRedisSerializer());
				//redis中设置订单过期时间1分钟
				redisTemplate.opsForValue().set("order:" + orders.getId(), orders, 1, TimeUnit.MINUTES);
				/**
				 * 给订单创建副本，副本的过期时间长于原订单
				 * redis过期后触发过期事件时，redis数据已经过期，此时只能拿到key，拿不到value。
				 * 而过期事件需要回退商品库存，必须拿到value即订单详情，才能拿到商品数据，进行回退操作
				 * 我们保存一个订单副本，过期时间长于原订单，此时就可以通过副本拿到原订单数据
				 */
				redisTemplate.opsForValue().set(orders.getId() + "_copy", orders, 2, TimeUnit.MINUTES);
				
				return orders;
				
			} finally {
				//不论否抛出异常，都释放锁
				redissonLock.unlock(lockKey);
			}
		}else {
			return null;
		}
	}
	
	/**
	 * 根据id查询秒杀订单
	 *
	 * @param id 订单id
	 * @return
	 */
	public Orders findOrder(String id) {
		return (Orders) redisTemplate.opsForValue().get("order:" + id);
	}
	
	/**
	 * 支付秒杀订单
	 *
	 * @param orderId 订单id
	 * @return
	 */
	public Orders pay(String orderId) {
		// 1.查询订单
		Orders orders = findOrder(orderId);
		if (orders == null) {
			throw new BusException(CodeEnum.ORDER_EXPIRED_ERROR);
		}
		
		//2.修改订单状态
		orders.setStatus(2);
		orders.setPaymentTime(new Date());
		orders.setPaymentType(2); // 支付宝支付
		
		
		//3.删除redis订单和复制订单
		redisTemplate.delete("order:" + orderId);
		redisTemplate.delete(orderId+"_copy");
		return orders;
	}
	
	/**
	 * 将一个秒杀商品保存到redis中
	 * @param seckillGoods 秒杀商品对象
	 */
	public void addRedisSeckillGoods(SeckillGoods seckillGoods) {
		redisTemplate.boundHashOps(SECKILL_GOODS).put(seckillGoods.getGoodsId(),seckillGoods);
		//将正在秒杀的商品保存到布隆过滤器
		bitMapBloomFilter.add(seckillGoods.getGoodsId().toString());
	}
}
